package com.topvision.myviewtest.rxjava;

import android.os.Bundle;
import android.support.annotation.Nullable;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.widget.Toast;

import com.topvision.myviewtest.rxjava.bean.ZipBean;
import com.topvision.myviewtest.rxjava.learn.Api;
import com.topvision.myviewtest.rxjava.learn.RetrofitProvider;
import com.topvision.myviewtest.rxjava.learn.entity.LoginRequest;
import com.topvision.myviewtest.rxjava.learn.entity.LoginResponse;
import com.topvision.myviewtest.rxjava.learn.entity.RegisterRequest;
import com.topvision.myviewtest.rxjava.learn.entity.RegisterResponse;
import com.topvision.myviewtest.rxjava.learn.entity.UserBaseInfoRequest;
import com.topvision.myviewtest.rxjava.learn.entity.UserBaseInfoResponse;
import com.topvision.myviewtest.rxjava.learn.entity.UserExtraInfoRequest;
import com.topvision.myviewtest.rxjava.learn.entity.UserExtraInfoResponse;
import com.topvision.myviewtest.rxjava.learn.entity.UserInfo;
import com.topvision.myviewtest.utils.LogUtils;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.io.InterruptedIOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.schedulers.Schedulers;

/**
 * Comment: //TODO
 * Date: 2017-11-08 16:01
 *
 * @author lucy
 */

public class RxJavaLearnActivity extends AppCompatActivity {
    private final String TAG = getClass().getSimpleName();

    static {
        RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                if (throwable instanceof InterruptedIOException) {
                    System.out.println("InterruptedIOException");
                }
            }
        });
    }

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
//        rxLearn();
//        rxLearnChain();
//        rxDispose();
//        rxConsumer();
//        rxThreadSwtch();
//        rxThreadTest();
//        rxThreadTest2();
//        rxRetrofit();
//        rxMap();
//        rxFlatMap();
//        rxConcatMap();
//        rxFlatMapPractice();
//        rxZipSyn();
//        rxZipAsyn();
//        rxZipPractice();
//        rxBackpressureTest();
        rxFlowble();
    }

    /**
     * 分解操作（写法）
     */
    private void rxLearn() {
        /* 创建一个Observerble */
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onNext(4);
                emitter.onComplete();
            }
        });
        /* 创建一个Observer */
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable disposable) {
                LogUtils.d(TAG, "subscribe");
            }

            @Override
            public void onNext(Integer value) {
                LogUtils.d(TAG, String.valueOf(value));
            }

            @Override
            public void onError(Throwable throwable) {
                LogUtils.d(TAG, "error");
            }

            @Override
            public void onComplete() {
                LogUtils.d(TAG, "complete");
            }
        };

        /* 建立连接 */
        observable.subscribe(observer);
    }

    /**
     * 链式编程（写法）
     */
    private void rxLearnChain() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onNext(4);
                emitter.onNext(5);
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable disposable) {
                LogUtils.d(TAG, "subscribe");
            }

            @Override
            public void onNext(Integer value) {
                LogUtils.d(TAG, String.valueOf(value));
            }

            @Override
            public void onError(Throwable throwable) {
                LogUtils.d(TAG, "error");
            }

            @Override
            public void onComplete() {
                LogUtils.d(TAG, "complete");
            }
        });
    }

    /**
     * dispoter
     */
    private void rxDispose() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                LogUtils.d(TAG, "emit 1");
                emitter.onNext(1);
                LogUtils.d(TAG, "emit 2");
                emitter.onNext(2);
                LogUtils.d(TAG, "emit 3");
                emitter.onNext(3);
                LogUtils.d(TAG, "emit complete");
                emitter.onComplete();
                LogUtils.d(TAG, "emit 4");
                emitter.onNext(4);
            }
        }).subscribe(new Observer<Integer>() {
            private Disposable mDisposable;
            private int i;
            private static final int DISPOSE_VALUE = 2;

            @Override
            public void onSubscribe(Disposable d) {
                LogUtils.d(TAG, "subscribe");
                mDisposable = d;
            }

            @Override
            public void onNext(Integer value) {
                LogUtils.d(TAG, "onNext: " + value);
                i++;
                if (i == DISPOSE_VALUE) {
                    Log.d(TAG, "dispose");
                    mDisposable.dispose();
                    Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                LogUtils.d(TAG, "error");
            }

            @Override
            public void onComplete() {
                LogUtils.d(TAG, "complete");
            }
        });
    }

    /**
     * 只关心onNext方法
     */
    private void rxConsumer() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                LogUtils.d(TAG, String.valueOf(integer));
            }
        });
    }

    /**
     * 线程切换
     */
    private void rxThreadSwtch() {
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                LogUtils.d(TAG, Thread.currentThread().getName());
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
            }
        });
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                LogUtils.d(TAG, "onSubscribe");
            }

            @Override
            public void onNext(Integer value) {
                LogUtils.d(TAG, "onNext: " + value + ", " + Thread.currentThread().getName());
            }

            @Override
            public void onError(Throwable e) {
                LogUtils.d(TAG, "onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d(TAG, "onComplete");
            }
        };
        observable
                //子线程执行
                .subscribeOn(Schedulers.newThread())
                //切换到主线程执行
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(observer);
        //结果
//        11-09 13:53:40.433 6625-6625/com.topvision.myviewtest D/RxJavaLearnActivity: onSubscribe
//        11-09 13:53:40.434 6625-6656/com.topvision.myviewtest D/RxJavaLearnActivity: RxNewThreadScheduler-1
//        11-09 13:53:40.495 6625-6625/com.topvision.myviewtest D/RxJavaLearnActivity: onNext1, main
//        11-09 13:53:40.495 6625-6625/com.topvision.myviewtest D/RxJavaLearnActivity: onNext2, main
//        11-09 13:53:40.495 6625-6625/com.topvision.myviewtest D/RxJavaLearnActivity: onNext3, main
//        11-09 13:53:40.495 6625-6625/com.topvision.myviewtest D/RxJavaLearnActivity: onComplete
    }

    /**
     * 测试线程切换的特性
     */
    private void rxThreadTest() {
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                LogUtils.d(TAG, Thread.currentThread().getName());
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
            }
        });
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                LogUtils.d(TAG, "onSubscribe");
            }

            @Override
            public void onNext(Integer value) {
                LogUtils.d(TAG, "onNext: " + value + ", " + Thread.currentThread().getName());
            }

            @Override
            public void onError(Throwable e) {
                LogUtils.d(TAG, "onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d(TAG, "onComplete");
            }
        };
        observable
                //指定被观察者（Observable）的执行线程
                .subscribeOn(Schedulers.newThread())
                //第二次指定被观察者（Observable）的执行线程
                .subscribeOn(AndroidSchedulers.mainThread())
                //指定观察者（Observer）的执行线程
                .observeOn(Schedulers.newThread())
                //第二次指定观察者（Observer）的执行线程
                .observeOn(AndroidSchedulers.mainThread())
                //订阅（绑定）
                .subscribe(observer);

        //执行结果
//        11-09 14:01:59.457 9581-9581/com.topvision.myviewtest D/RxJavaLearnActivity: onSubscribe
//        11-09 14:01:59.475 9581-10582/com.topvision.myviewtest D/RxJavaLearnActivity: RxNewThreadScheduler-5
//        11-09 14:01:59.490 9581-9581/com.topvision.myviewtest D/RxJavaLearnActivity: onNext: 1, main
//        11-09 14:01:59.490 9581-9581/com.topvision.myviewtest D/RxJavaLearnActivity: onNext: 2, main
//        11-09 14:01:59.490 9581-9581/com.topvision.myviewtest D/RxJavaLearnActivity: onNext: 3, main
//        11-09 14:01:59.490 9581-9581/com.topvision.myviewtest D/RxJavaLearnActivity: onComplete
        /* 被观察者（Observable）只有第一次设置线程有效，之后无效； */
        /* 观察者（Observer）每调用一次observeOn（）设置线程均会切换到该线程 */
    }

    private void rxThreadTest2() {
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                LogUtils.d(TAG, Thread.currentThread().getName());
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
            }
        });

        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                LogUtils.d(TAG, "onSubscribe");
            }

            @Override
            public void onNext(Integer value) {
                LogUtils.d(TAG, "onNext: " + value + ", " + Thread.currentThread().getName());
            }

            @Override
            public void onError(Throwable e) {
                LogUtils.d(TAG, "onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d(TAG, "onComplete");
            }
        };

        observable
                .subscribeOn(Schedulers.newThread())
                .subscribeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        LogUtils.d(TAG, "After observeOn(mainThread), current thread is: " + Thread.currentThread().getName());
                    }
                })
                .observeOn(Schedulers.io())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        LogUtils.d(TAG, "After observeOn(io), current thread is : " + Thread.currentThread().getName());
                    }
                })
                .subscribe(observer);

        /* 执行结果 */
//        11-13 10:53:48.081 20824-20824/com.topvision.myviewtest D/RxJavaLearnActivity: onSubscribe
//        11-13 10:53:48.133 20824-20872/com.topvision.myviewtest D/RxJavaLearnActivity: RxNewThreadScheduler-1
//        11-13 10:53:48.133 20824-20872/com.topvision.myviewtest D/RxJavaLearnActivity: After observeOn(mainThread), current thread is: RxNewThreadScheduler-1
//        11-13 10:53:48.133 20824-20872/com.topvision.myviewtest D/RxJavaLearnActivity: After observeOn(mainThread), current thread is: RxNewThreadScheduler-1
//        11-13 10:53:48.133 20824-20872/com.topvision.myviewtest D/RxJavaLearnActivity: After observeOn(mainThread), current thread is: RxNewThreadScheduler-1
//        11-13 10:53:48.133 20824-20873/com.topvision.myviewtest D/RxJavaLearnActivity: After observeOn(io), current thread is : RxCachedThreadScheduler-1
//        11-13 10:53:48.133 20824-20873/com.topvision.myviewtest D/RxJavaLearnActivity: onNext: 1, RxCachedThreadScheduler-1
//        11-13 10:53:48.134 20824-20873/com.topvision.myviewtest D/RxJavaLearnActivity: After observeOn(io), current thread is : RxCachedThreadScheduler-1
//        11-13 10:53:48.134 20824-20873/com.topvision.myviewtest D/RxJavaLearnActivity: onNext: 2, RxCachedThreadScheduler-1
//        11-13 10:53:48.134 20824-20873/com.topvision.myviewtest D/RxJavaLearnActivity: After observeOn(io), current thread is : RxCachedThreadScheduler-1
//        11-13 10:53:48.134 20824-20873/com.topvision.myviewtest D/RxJavaLearnActivity: onNext: 3, RxCachedThreadScheduler-1
//        11-13 10:53:48.134 20824-20873/com.topvision.myviewtest D/RxJavaLearnActivity: onComplete

        /*
        chedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
        Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
        Schedulers.newThread() 代表一个常规的新线程
        AndroidSchedulers.mainThread() 代表Android的主线程
        */
    }

    /**
     * retrofit的使用
     */
    private void rxRetrofit() {
        Api api = RetrofitProvider.get().create(Api.class);
        api.login(new LoginRequest())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<LoginResponse>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(LoginResponse value) {

                    }

                    @Override
                    public void onError(Throwable e) {
                        Toast.makeText(RxJavaLearnActivity.this, "登录失败", Toast.LENGTH_SHORT).show();
                    }

                    @Override
                    public void onComplete() {
                        Toast.makeText(RxJavaLearnActivity.this, "登录成功", Toast.LENGTH_SHORT).show();
                    }
                });

    }

    /**
     * Map操作符
     */
    private void rxMap() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                LogUtils.d(TAG, "收到了Integer:" + integer);
                return String.valueOf(integer);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                LogUtils.d(TAG, "收到了String:" + s);
            }
        });
        /* 执行结果 */
//        11-13 17:35:55.074 13995-13995/? D/RxJavaLearnActivity: 收到了Integer:1
//        11-13 17:35:55.074 13995-13995/? D/RxJavaLearnActivity: 收到了String:1
//        11-13 17:35:55.074 13995-13995/? D/RxJavaLearnActivity: 收到了Integer:2
//        11-13 17:35:55.074 13995-13995/? D/RxJavaLearnActivity: 收到了String:2
//        11-13 17:35:55.074 13995-13995/? D/RxJavaLearnActivity: 收到了Integer:3
//        11-13 17:35:55.074 13995-13995/? D/RxJavaLearnActivity: 收到了String:3
    }

    private void rxFlatMap() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
//                return Observable.fromIterable(list);
                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                LogUtils.d(TAG, "收到了:" + s);
            }
        });
        /* 执行结果 */
//        11-13 17:42:09.821 21619-21652/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 1
//        11-13 17:42:09.821 21619-21652/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 2
//        11-13 17:42:09.821 21619-21652/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 2
//        11-13 17:42:09.821 21619-21652/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 2
//        11-13 17:42:09.821 21619-21652/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 3
//        11-13 17:42:09.821 21619-21652/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 3
//        11-13 17:42:09.821 21619-21652/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 3
//        11-13 17:42:09.821 21619-21652/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 1
//        11-13 17:42:09.821 21619-21652/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 1
        /* flatMap并不保证事件的顺序 */
    }

    /**
     * concatMap操作符
     */
    private void rxConcatMap() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
//                return Observable.fromIterable(list);
                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                LogUtils.d(TAG, "收到了:" + s);
            }
        });
        /* 执行结果 */
//        11-13 17:48:44.038 28784-28816/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 1
//        11-13 17:48:44.038 28784-28816/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 1
//        11-13 17:48:44.038 28784-28816/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 1
//        11-13 17:48:44.049 28784-28823/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 2
//        11-13 17:48:44.049 28784-28823/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 2
//        11-13 17:48:44.049 28784-28823/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 2
//        11-13 17:48:44.059 28784-28824/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 3
//        11-13 17:48:44.060 28784-28824/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 3
//        11-13 17:48:44.060 28784-28824/com.topvision.myviewtest D/RxJavaLearnActivity: 收到了:I am value 3
        /* concatMap可以保证事件的顺序 */
    }

    /**
     * flatMap实践（迷茫中...待理解）
     */
    private void rxFlatMapPractice() {
        final Api api = RetrofitProvider.get().create(Api.class);
        api.register(new RegisterRequest())
                .subscribeOn(Schedulers.io())
                .doOnNext(new Consumer<RegisterResponse>() {
                    @Override
                    public void accept(RegisterResponse registerResponse) throws Exception {
                        //根据注册的响应结果去做一些操作
                    }
                })
                .observeOn(Schedulers.io())
                .flatMap(new Function<RegisterResponse, ObservableSource<LoginResponse>>() {
                    @Override
                    public ObservableSource<LoginResponse> apply(RegisterResponse registerResponse) throws Exception {
                        //注册后一些操作
                        return api.login(new LoginRequest());
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<LoginResponse>() {
                    @Override
                    public void accept(LoginResponse loginResponse) throws Exception {
                        Toast.makeText(RxJavaLearnActivity.this, "登录成功", Toast.LENGTH_SHORT).show();
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Toast.makeText(RxJavaLearnActivity.this, "登录失败", Toast.LENGTH_SHORT).show();
                    }
                });
    }

    /**
     * zip操作符
     * 它只发射与发射数据项最少的那个Observable一样多的数据，发完之后Observer就不会收到剩余的事件了.
     * 组合的顺序是严格按照事件发送的顺利
     *
     */
    private void rxZipSyn() {
        //当前操作都在MainThread
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                LogUtils.d(TAG, String.valueOf(1));
                e.onNext(1);
                LogUtils.d(TAG, String.valueOf(2));
                e.onNext(2);
                LogUtils.d(TAG, String.valueOf(3));
                e.onNext(3);
                e.onComplete();
            }
        });
        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                LogUtils.d(TAG, "A");
                e.onNext("A");
                LogUtils.d(TAG, "B");
                e.onNext("B");
                LogUtils.d(TAG, "C");
                e.onNext("C");
                LogUtils.d(TAG, "D");
                e.onNext("D");
                e.onComplete();
            }
        });
        Observable.zip(observable1, observable2, new BiFunction<Integer, String, ZipBean>() {

            @Override
            public ZipBean apply(Integer integer, String s) throws Exception {
                return new ZipBean(s, integer);
            }
        }).subscribe(new Observer<ZipBean>() {
            @Override
            public void onSubscribe(Disposable d) {
                LogUtils.d(TAG, "onSubscribe");
            }

            @Override
            public void onNext(ZipBean value) {
                LogUtils.d(TAG, "String:" + value.getString() + ", Integer:" + value.getInteger());
            }

            @Override
            public void onError(Throwable e) {
                LogUtils.d(TAG, "onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d(TAG, "onComplete");
            }
        });
        /* 执行结果 */
//        11-13 18:10:50.336 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: onSubscribe
//        11-13 18:10:50.336 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: 1
//        11-13 18:10:50.336 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: 2
//        11-13 18:10:50.336 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: 3
//        11-13 18:10:50.336 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: A
//        11-13 18:10:50.337 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: String:A, Integer:1
//        11-13 18:10:50.337 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: B
//        11-13 18:10:50.337 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: String:B, Integer:2
//        11-13 18:10:50.337 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: C
//        11-13 18:10:50.337 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: String:C, Integer:3
//        11-13 18:10:50.337 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: onComplete
//        11-13 18:10:50.337 23544-23544/com.topvision.myviewtest D/RxJavaLearnActivity: D
        /* 同一线程会执行完第一个操作才执行第二个操作 */
    }

    private void rxZipAsyn() {
        //这里采用异步操作
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                LogUtils.d(TAG, String.valueOf(1));
                e.onNext(1);
                Thread.sleep(10);
                LogUtils.d(TAG, String.valueOf(2));
                e.onNext(2);
                Thread.sleep(10);
                LogUtils.d(TAG, String.valueOf(3));
                e.onNext(3);
                Thread.sleep(10);
                e.onComplete();
            }
        }).subscribeOn(Schedulers.io());
        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                LogUtils.d(TAG, "A");
                e.onNext("A");
                Thread.sleep(10);
                LogUtils.d(TAG, "B");
                e.onNext("B");
                Thread.sleep(10);
                LogUtils.d(TAG, "C");
                e.onNext("C");
                Thread.sleep(10);
                LogUtils.d(TAG, "D");
                e.onNext("D");
                Thread.sleep(10);
                e.onComplete();
            }
        }).subscribeOn(Schedulers.io());
        Observable.zip(observable1, observable2, new BiFunction<Integer, String, ZipBean>() {

            @Override
            public ZipBean apply(Integer integer, String s) throws Exception {
                return new ZipBean(s, integer);
            }
        }).subscribe(new Observer<ZipBean>() {
            @Override
            public void onSubscribe(Disposable d) {
                LogUtils.d(TAG, "onSubscribe");
            }

            @Override
            public void onNext(ZipBean value) {
                LogUtils.d(TAG, "String:" + value.getString() + ", Integer:" + value.getInteger());
            }

            @Override
            public void onError(Throwable e) {
                LogUtils.d(TAG, "onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d(TAG, "onComplete");
            }
        });
        /* 执行结果 */
//        11-14 09:47:03.553 16825-16825/? D/RxJavaLearnActivity: onSubscribe
//        11-14 09:47:03.554 16825-16878/? D/RxJavaLearnActivity: 1
//        11-14 09:47:03.555 16825-16879/? D/RxJavaLearnActivity: A
//        11-14 09:47:03.555 16825-16879/? D/RxJavaLearnActivity: String:A, Integer:1
//        11-14 09:47:03.565 16825-16878/? D/RxJavaLearnActivity: 2
//        11-14 09:47:03.565 16825-16879/? D/RxJavaLearnActivity: B
//        11-14 09:47:03.565 16825-16879/? D/RxJavaLearnActivity: String:B, Integer:2
//        11-14 09:47:03.575 16825-16878/? D/RxJavaLearnActivity: 3
//        11-14 09:47:03.575 16825-16879/? D/RxJavaLearnActivity: C
//        11-14 09:47:03.575 16825-16879/? D/RxJavaLearnActivity: String:C, Integer:3
//        11-14 09:47:03.585 16825-16878/? D/RxJavaLearnActivity: onComplete
        /* 按顺序1-1合成 */
    }

    /**
     * zip操作符实践
     */
    private void rxZipPractice() {
        Api api = RetrofitProvider.get().create(Api.class);
        Observable<UserBaseInfoResponse> userBaseInfo = api.getUserBaseInfo(new UserBaseInfoRequest());

        Observable<UserExtraInfoResponse> userExtraInfo = api.getUserExtraInfo(new UserExtraInfoRequest());

        Observable.zip(userBaseInfo, userExtraInfo, new BiFunction<UserBaseInfoResponse, UserExtraInfoResponse, UserInfo>() {
            @Override
            public UserInfo apply(UserBaseInfoResponse userBaseInfoResponse, UserExtraInfoResponse userExtraInfoResponse) throws Exception {
                return new UserInfo(userBaseInfoResponse, userExtraInfoResponse);
            }
        }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<UserInfo>() {
            @Override
            public void accept(UserInfo userInfo) throws Exception {
                //正常操作
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                //异常操作
            }
        });
    }

    private void rxBackpressureTest() {
        //1.采用无延时无complete发送，导致内存逐渐提升，最后爆炸OOM
        //2.采用延时发送，内存控制住
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                //第一个observable采用无限循环发送数据
                while (true) {
                    LogUtils.d(TAG, String.valueOf(1));
                    e.onNext(1);
                }
            }
        }).subscribeOn(Schedulers.io()).sample(2, TimeUnit.SECONDS);

        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                //第二个observable只发送2个数据
                e.onNext("A");
                e.onNext("B");
            }
        }).subscribeOn(Schedulers.io());
        Observable.zip(observable1, observable2, new BiFunction<Integer, String, ZipBean>() {
            @Override
            public ZipBean apply(Integer integer, String s) throws Exception {
                return new ZipBean(s, integer);
            }
        }).subscribeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<ZipBean>() {
            @Override
            public void accept(ZipBean zipBean) throws Exception {
                //正常操作
                LogUtils.d(TAG, "String:" + zipBean.getString() + ", Integer:" + zipBean.getInteger());
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                //异常操作
                LogUtils.e(TAG, throwable.toString());
            }
        });
    }

    /**
     *     Flowble和Subscriber 有背压Backpressure
     *     Observable和Observer是无背压Backpressure的
     */
    private void rxFlowble() {
        //lowble和Subscriber
        //Observable和Observer
        //两者的区别就在与有无被压，使用方式还是一样的
        Flowable<Integer> flowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                LogUtils.d(TAG,"Emitter:"+1);
                e.onNext(2);
                LogUtils.d(TAG,"Emitter:"+2);
                e.onNext(3);
                LogUtils.d(TAG,"Emitter:"+3);
                e.onComplete();
                LogUtils.d(TAG,"Emitter:onComplete");
            }
            /* 增加了一个参数
               BackpressureStrategy.ERROR Flowable只有一个128大小的buffer
               BackpressureStrategy.BUFFER Flowable有一个无限大小的buffer
               BackpressureStrategy.DROP Flowable只会保存前128个数据，其他内容直接丢弃，只有Subscriber请求完128个数据后，再保存另128个数据
               BackpressureStrategy.LATEST 与Drop不同, Latest总是能获取到最后最新的事件


                */
        }, BackpressureStrategy.ERROR);

        Subscriber<Integer> subscriber = new Subscriber<Integer>() {

            @Override
            public void onSubscribe(Subscription s) {
                LogUtils.d(TAG,"onSubscribe");
                //Flowber必须使用request()方法请求数据
                s.request(Long.MAX_VALUE);
                /* Subscription只有2个方法request()和cancel() */
            }

            @Override
            public void onNext(Integer integer) {
                LogUtils.d(TAG,"onNext:"+integer);
            }

            @Override
            public void onError(Throwable t) {
                LogUtils.d(TAG,"onError:"+t.toString());
            }

            @Override
            public void onComplete() {
                LogUtils.d(TAG,"onComplete");
            }
        };

        flowable.subscribe(subscriber);

        /* 执行结果 */
//        11-17 13:39:25.774 18506-18506/com.topvision.myviewtest D/RxJavaLearnActivity: onSubscribe
//        11-17 13:39:25.775 18506-18506/com.topvision.myviewtest D/RxJavaLearnActivity: onError:
//        io.reactivex.exceptions.MissingBackpressureException: create: could not emit value due to lack of requests
//        11-17 13:39:25.775 18506-18506/com.topvision.myviewtest D/RxJavaLearnActivity: Emitter:1
//        11-17 13:39:25.775 18506-18506/com.topvision.myviewtest D/RxJavaLearnActivity: Emitter:2
//        11-17 13:39:25.775 18506-18506/com.topvision.myviewtest D/RxJavaLearnActivity: Emitter:3
//        11-17 13:39:25.775 18506-18506/com.topvision.myviewtest D/RxJavaLearnActivity: Emitter:onComplete
        /* 出现了MissingBackpressureException错误 */
        /* Flowable需要我们自己去拉取Flowber发送的数据，所以要在onSubscribe中添加s.request(Long.MAX_VALUE) */

        /* 之后的执行结果 */
//        11-17 13:38:08.206 17041-17041/com.topvision.myviewtest D/RxJavaLearnActivity: onSubscribe
//        11-17 13:38:08.206 17041-17041/com.topvision.myviewtest D/RxJavaLearnActivity: onNext:1
//        11-17 13:38:08.206 17041-17041/com.topvision.myviewtest D/RxJavaLearnActivity: Emitter:1
//        11-17 13:38:08.206 17041-17041/com.topvision.myviewtest D/RxJavaLearnActivity: onNext:2
//        11-17 13:38:08.206 17041-17041/com.topvision.myviewtest D/RxJavaLearnActivity: Emitter:2
//        11-17 13:38:08.206 17041-17041/com.topvision.myviewtest D/RxJavaLearnActivity: onNext:3
//        11-17 13:38:08.206 17041-17041/com.topvision.myviewtest D/RxJavaLearnActivity: Emitter:3
//        11-17 13:38:08.206 17041-17041/com.topvision.myviewtest D/RxJavaLearnActivity: onComplete
//        11-17 13:38:08.206 17041-17041/com.topvision.myviewtest D/RxJavaLearnActivity: Emitter:onComplete

        /* Flowable里默认有一个大小为128的buffer.在不同的线程中时, Flowable就会先把事件发送到这个buffer中,
         因此, Subscriber虽然没有调用request, 但是Flowable在buffer中保存着这些事件, 只有当Subscriber调用
         request时, 才从buffer里取出事件发给Subscriber. */
    }

    /**
     * interval操作符
     */
    private void rxInterval() {
        /*interval操作符发送Long型的事件, 从0开始, 每隔指定的时间就把数字加1并发送出来,
        在这个例子里, 我们让它每隔1毫秒就发送一次事件, 在下游延时1秒去接收处理, 不用猜也知道结果是什么*/
        Flowable.interval(1, TimeUnit.SECONDS)
                .onBackpressureBuffer()
//                .onBackpressureDrop()
//                .onBackpressureLatest()
                .subscribeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Long>() {
                               @Override
                               public void onSubscribe(Subscription s) {
                                   LogUtils.d(TAG, "onSubscribe");
                                   s.request(Long.MAX_VALUE);
                               }

                               @Override
                               public void onNext(Long aLong) {
                                   LogUtils.d(TAG, "onNext: " + aLong);
                                   try {
                                       //延时1秒
                                       Thread.sleep(1000);
                                   } catch (InterruptedException e) {
                                       e.printStackTrace();
                                   }
                               }

                               @Override
                               public void onError(Throwable t) {
                                   LogUtils.w(TAG, "onError: " + t.toString());
                               }

                               @Override
                               public void onComplete() {
                                   LogUtils.d(TAG, "onComplete");
                               }
                           });
        /* 一运行就抛出了MissingBackpressureException异常, 提醒我们发太多了, 那么怎么办呢, 这个又不是我们自己创建的FLowable */
        /* RxJava给我们提供了其他的方法: */
//        onBackpressureBuffer()
//        onBackpressureDrop()
//        onBackpressureLatest()

    }
}
